package com.poetic.order.function;

import com.poetic.order.domain.OrderEvent;
import com.poetic.order.domain.OrderResult;
import com.poetic.order.domain.ReceiptEvent;
import lombok.val;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * <pre>
 *  TODO
 * Created by lianghuikun on 2020-09-16.
 * </pre>
 *
 * @author lianghuikun
 */
public class TxPayMatch extends CoProcessFunction<OrderEvent, ReceiptEvent, Tuple2<OrderEvent, ReceiptEvent>> {

    private OutputTag<OrderEvent> unmatchedPays;
    private OutputTag<ReceiptEvent> unmatchedReceipts;

    public TxPayMatch(OutputTag<OrderEvent> unmatchedPays, OutputTag<ReceiptEvent> unmatchedReceipts) {
        this.unmatchedPays = unmatchedPays;
        this.unmatchedReceipts = unmatchedReceipts;
    }

    // 定义状态来保存已经到达的订单支付事件和到账事件
    ValueState<OrderEvent> payState;
    // 定义状态，用于保存定时器的时间戳（用于删除）
    ValueState<ReceiptEvent> receiptState;

    @Override
    public void open(Configuration parameters) throws Exception {
        payState = getRuntimeContext().getState(new ValueStateDescriptor<OrderEvent>("payState",
                TypeInformation.of(new TypeHint<OrderEvent>() {
                })));
        receiptState = getRuntimeContext().getState(new ValueStateDescriptor<ReceiptEvent>("receiptState",
                TypeInformation.of(new TypeHint<ReceiptEvent>() {
                })));
    }

    // 订单支付事件数据的处理
    @Override
    public void processElement1(OrderEvent pay, Context ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
        // 判断有没有对应的到账事件
        ReceiptEvent receipt = receiptState.value();
        if (receipt != null) {
            // 如果已经有receipt，在主流输出匹配信息，清空状态
            out.collect(Tuple2.of(pay, receipt));
            receiptState.clear();
        } else {
            // 如果还没到，那么把pay存入状态，并且注册一个定时器等待
            payState.update(pay);
            ctx.timerService().registerEventTimeTimer(pay.getEventTime() * 1000L + 5000L);
        }
    }

    // 到账事件的处理
    @Override
    public void processElement2(ReceiptEvent receipt, Context ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
        // 同样的处理流程
        OrderEvent pay = payState.value();
        if (pay != null) {
            out.collect(Tuple2.of(pay, receipt));
            payState.clear();
        } else {
            receiptState.update(receipt);
            ctx.timerService().registerEventTimeTimer(receipt.getEventTime() * 1000L + 5000L);
        }
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<Tuple2<OrderEvent, ReceiptEvent>> out) throws Exception {
        // 到时间了，如果还没有收到某个事件，那么输出报警信息
        if( payState.value() != null ){
            // recipt没来，输出pay到侧输出流
            ctx.output(unmatchedPays, payState.value());
        }
        if( receiptState.value() != null ){
            ctx.output(unmatchedReceipts, receiptState.value());
        }
        payState.clear();
        receiptState.clear();
    }
}
